//! 解耦的连接处理器
//! 
//! 核心设计原则：
//! 1. Tokio只出现在网络层
//! 2. 协议层保持同步，不依赖任何异步运行时
//! 3. 通过标准库的mpsc队列桥接网络层和协议层
//! 4. 协议层可以独立测试，不依赖Tokio

use std::sync::{Arc, Mutex, mpsc};
use std::thread::{self, JoinHandle};
use std::time::Duration;
use std::collections::VecDeque;

use crate::protocol::RespParser;
use crate::storage::{Command, CommandResult, SingleThreadCommandDispatcher};
use crate::event::ConnectionId;

/// 连接命令（网络层发送给协议层）
#[derive(Debug)]
pub enum ConnectionCommand {
    /// 新数据到达
    DataReceived(Vec<u8>),
    /// 连接关闭
    ConnectionClosed,
    /// 停止处理
    Shutdown,
}

/// 连接响应（协议层发送给网络层）
#[derive(Debug)]
pub enum ConnectionResponse {
    /// 需要发送的响应数据
    SendData(Vec<u8>),
    /// 连接处理完成
    Finished,
    /// 错误信息
    Error(String),
}

/// 解耦的连接处理器
/// 
/// 核心特性：
/// 1. 协议层完全同步，不依赖Tokio
/// 2. 使用标准库mpsc队列与网络层通信
/// 3. 独立的协议处理线程
/// 4. 网络层和协议层完全解耦
pub struct DecoupledConnectionHandler {
    /// 连接ID
    connection_id: ConnectionId,
    /// 发送命令到协议层
    command_sender: mpsc::Sender<ConnectionCommand>,
    /// 接收协议层的响应
    response_receiver: mpsc::Receiver<ConnectionResponse>,
    /// 协议处理线程句柄
    protocol_thread: Option<JoinHandle<()>>,
}

impl DecoupledConnectionHandler {
    /// 创建新的解耦连接处理器
    pub fn new(
        connection_id: ConnectionId,
        storage_dispatcher: Arc<Mutex<SingleThreadCommandDispatcher>>,
    ) -> Self {
        // 创建标准库的mpsc通道（不依赖Tokio）
        let (command_sender, command_receiver) = mpsc::channel();
        let (response_sender, response_receiver) = mpsc::channel();

        // 启动协议处理线程（纯同步，不依赖Tokio）
        let protocol_thread = Self::spawn_protocol_thread(
            connection_id,
            command_receiver,
            response_sender,
            storage_dispatcher,
        );

        Self {
            connection_id,
            command_sender,
            response_receiver,
            protocol_thread: Some(protocol_thread),
        }
    }

    /// 网络层调用：发送新数据到协议层
    pub fn handle_data(&self, data: Vec<u8>) -> Result<(), String> {
        self.command_sender.send(ConnectionCommand::DataReceived(data))
            .map_err(|_| "协议层已关闭".to_string())
    }

    /// 网络层调用：通知连接关闭
    pub fn handle_close(&self) -> Result<(), String> {
        self.command_sender.send(ConnectionCommand::ConnectionClosed)
            .map_err(|_| "协议层已关闭".to_string())
    }

    /// 网络层调用：获取待发送的响应（非阻塞）
    pub fn try_get_response(&self) -> Option<ConnectionResponse> {
        self.response_receiver.try_recv().ok()
    }

    /// 网络层调用：获取待发送的响应（阻塞，带超时）
    pub fn get_response_timeout(&self, timeout: Duration) -> Option<ConnectionResponse> {
        self.response_receiver.recv_timeout(timeout).ok()
    }

    /// 停止连接处理器
    pub fn stop(&mut self) -> Result<(), String> {
        // 发送停止命令
        let _ = self.command_sender.send(ConnectionCommand::Shutdown);

        // 等待协议线程结束
        if let Some(handle) = self.protocol_thread.take() {
            handle.join()
                .map_err(|_| "协议线程结束失败".to_string())?;
        }

        println!("✅ 连接 {} 解耦处理器已停止", self.connection_id);
        Ok(())
    }

    /// 启动协议处理线程（纯同步，不依赖Tokio）
    fn spawn_protocol_thread(
        connection_id: ConnectionId,
        command_receiver: mpsc::Receiver<ConnectionCommand>,
        response_sender: mpsc::Sender<ConnectionResponse>,
        storage_dispatcher: Arc<Mutex<SingleThreadCommandDispatcher>>,
    ) -> JoinHandle<()> {
        thread::Builder::new()
            .name(format!("protocol-{connection_id}"))
            .spawn(move || {
                Self::protocol_loop(
                    connection_id,
                    command_receiver,
                    response_sender,
                    storage_dispatcher,
                );
            })
            .expect("创建协议线程失败")
    }

    /// 协议处理循环（完全同步，不依赖任何异步运行时）
    fn protocol_loop(
        connection_id: ConnectionId,
        command_receiver: mpsc::Receiver<ConnectionCommand>,
        response_sender: mpsc::Sender<ConnectionResponse>,
        storage_dispatcher: Arc<Mutex<SingleThreadCommandDispatcher>>,
    ) {
        println!("🔄 连接 {connection_id} 协议处理循环开始（纯同步）");

        let mut resp_parser = RespParser::new();
        let mut pending_commands = VecDeque::new();

        // 主处理循环
        while let Ok(command) = command_receiver.recv() {
            match command {
                ConnectionCommand::DataReceived(data) => {
                    // 处理接收到的数据
                    Self::handle_received_data(
                        connection_id,
                        &data,
                        &mut resp_parser,
                        &mut pending_commands,
                        &storage_dispatcher,
                        &response_sender,
                    );
                }
                ConnectionCommand::ConnectionClosed => {
                    println!("📡 连接 {connection_id} 已关闭");
                    break;
                }
                ConnectionCommand::Shutdown => {
                    println!("🛑 连接 {connection_id} 收到停止命令");
                    break;
                }
            }
        }

        // 发送完成信号
        let _ = response_sender.send(ConnectionResponse::Finished);
        println!("✅ 连接 {connection_id} 协议处理循环结束");
    }

    /// 处理接收到的数据（纯同步）
    fn handle_received_data(
        connection_id: ConnectionId,
        data: &[u8],
        resp_parser: &mut RespParser,
        pending_commands: &mut VecDeque<Box<dyn Command>>,
        storage_dispatcher: &Arc<Mutex<SingleThreadCommandDispatcher>>,
        response_sender: &mpsc::Sender<ConnectionResponse>,
    ) {
        // 1. 将数据喂给RESP解析器
        resp_parser.feed_data(data);

        // 2. 解析所有可能的命令
        while let Ok(Some(resp_value)) = resp_parser.parse_value() {
            match resp_parser.resp_to_command(resp_value) {
                Ok(command) => {
                    pending_commands.push_back(command);
                }
                Err(e) => {
                    eprintln!("❌ 连接 {connection_id} 命令解析失败: {e:?}");
                    let error_response = b"-ERR Protocol error\r\n".to_vec();
                    let _ = response_sender.send(ConnectionResponse::SendData(error_response));
                }
            }
        }

        // 3. 处理所有待处理的命令
        while let Some(command) = pending_commands.pop_front() {
            Self::execute_command(
                connection_id,
                command,
                storage_dispatcher,
                response_sender,
            );
        }
    }

    /// 执行单个命令（纯同步）
    fn execute_command(
        connection_id: ConnectionId,
        command: Box<dyn Command>,
        storage_dispatcher: &Arc<Mutex<SingleThreadCommandDispatcher>>,
        response_sender: &mpsc::Sender<ConnectionResponse>,
    ) {
        let command_name = command.name().to_string();

        // 分发命令到存储层
        match storage_dispatcher.lock() {
            Ok(dispatcher) => {
                match dispatcher.dispatch_command(command) {
                    Ok(()) => {
                        // 模拟命令执行结果（实际应该从存储层获取）
                        let result = match command_name.as_str() {
                            "PING" => CommandResult::String("PONG".to_string()),
                            "SET" => CommandResult::Ok,
                            "GET" => CommandResult::String("mock_value".to_string()),
                            _ => CommandResult::Ok,
                        };

                        // 将结果转换为RESP格式并发送
                        let response_data = result.to_resp().into_bytes();
                        let _ = response_sender.send(ConnectionResponse::SendData(response_data));

                        println!("📝 连接 {connection_id} 命令 {command_name} 已处理");
                    }
                    Err(e) => {
                        eprintln!("❌ 连接 {connection_id} 命令 {command_name} 执行失败: {e}");
                        let error_response = format!("-ERR {e}\r\n").into_bytes();
                        let _ = response_sender.send(ConnectionResponse::SendData(error_response));
                    }
                }
            }
            Err(e) => {
                eprintln!("❌ 连接 {connection_id} 获取存储分发器失败: {e}");
                let error_response = b"-ERR Internal error\r\n".to_vec();
                let _ = response_sender.send(ConnectionResponse::SendData(error_response));
            }
        }
    }

    /// 获取连接ID
    pub fn connection_id(&self) -> ConnectionId {
        self.connection_id
    }
}

impl Drop for DecoupledConnectionHandler {
    fn drop(&mut self) {
        let _ = self.stop();
    }
}

/// 解耦的网络服务器接口
/// 
/// 网络层实现这个接口，协议层通过标准库的channel通信
pub trait NetworkServerInterface {
    /// 发送数据到指定连接
    fn send_to_connection(&self, connection_id: ConnectionId, data: Vec<u8>) -> Result<(), String>;
    
    /// 关闭指定连接
    fn close_connection(&self, connection_id: ConnectionId) -> Result<(), String>;
    
    /// 获取活跃连接数
    fn get_active_connections(&self) -> usize;
}

/// 解耦的协议服务器
/// 
/// 管理所有解耦的连接处理器，完全不依赖Tokio
pub struct DecoupledProtocolServer {
    /// 连接处理器映射
    connections: Arc<Mutex<std::collections::HashMap<ConnectionId, DecoupledConnectionHandler>>>,
    /// 存储分发器
    storage_dispatcher: Arc<Mutex<SingleThreadCommandDispatcher>>,
    /// 连接计数器
    connection_counter: Arc<Mutex<u64>>,
}

impl DecoupledProtocolServer {
    /// 创建新的解耦协议服务器
    pub fn new(storage_dispatcher: SingleThreadCommandDispatcher) -> Self {
        Self {
            connections: Arc::new(Mutex::new(std::collections::HashMap::new())),
            storage_dispatcher: Arc::new(Mutex::new(storage_dispatcher)),
            connection_counter: Arc::new(Mutex::new(0)),
        }
    }

    /// 创建新连接处理器
    pub fn create_connection(&self) -> Result<(ConnectionId, DecoupledConnectionHandler), String> {
        let connection_id = {
            let mut counter = self.connection_counter.lock().unwrap();
            *counter += 1;
            *counter
        };

        let handler = DecoupledConnectionHandler::new(
            connection_id,
            Arc::clone(&self.storage_dispatcher),
        );

        // 存储连接处理器
        {
            let mut connections = self.connections.lock().unwrap();
            connections.insert(connection_id, handler);
        }

        // 返回新的处理器实例
        let new_handler = DecoupledConnectionHandler::new(
            connection_id,
            Arc::clone(&self.storage_dispatcher),
        );

        println!("🌐 创建新连接处理器: ID={connection_id}");
        Ok((connection_id, new_handler))
    }

    /// 移除连接处理器
    pub fn remove_connection(&self, connection_id: ConnectionId) -> Result<(), String> {
        let mut connections = self.connections.lock().unwrap();
        if let Some(mut handler) = connections.remove(&connection_id) {
            handler.stop()?;
            println!("🗑️ 移除连接处理器: ID={connection_id}");
        }
        Ok(())
    }

    /// 获取活跃连接数
    pub fn get_active_connections_count(&self) -> usize {
        self.connections.lock().unwrap().len()
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn test_decoupled_connection_handler() {
        let storage_dispatcher = SingleThreadCommandDispatcher::new();
        let mut handler = DecoupledConnectionHandler::new(
            1,
            Arc::new(Mutex::new(storage_dispatcher)),
        );

        // 发送PING命令数据
        let ping_data = b"*1\r\n$4\r\nPING\r\n".to_vec();
        assert!(handler.handle_data(ping_data).is_ok());

        // 等待处理
        std::thread::sleep(Duration::from_millis(10));

        // 检查响应
        if let Some(ConnectionResponse::SendData(response)) = handler.try_get_response() {
            let response_str = String::from_utf8_lossy(&response);
            assert!(response_str.contains("PONG"));
        }

        // 停止处理器
        assert!(handler.stop().is_ok());
    }

    #[test]
    fn test_decoupled_protocol_server() {
        let storage_dispatcher = SingleThreadCommandDispatcher::new();
        let server = DecoupledProtocolServer::new(storage_dispatcher);

        // 创建连接
        let (conn_id, mut handler) = server.create_connection().unwrap();
        assert_eq!(server.get_active_connections_count(), 1);

        // 测试命令处理
        let set_data = b"*3\r\n$3\r\nSET\r\n$3\r\nkey\r\n$5\r\nvalue\r\n".to_vec();
        assert!(handler.handle_data(set_data).is_ok());

        // 等待处理
        std::thread::sleep(Duration::from_millis(10));

        // 检查响应
        if let Some(ConnectionResponse::SendData(response)) = handler.try_get_response() {
            let response_str = String::from_utf8_lossy(&response);
            assert!(response_str.contains("OK"));
        }

        // 清理
        assert!(handler.stop().is_ok());
        assert!(server.remove_connection(conn_id).is_ok());
        assert_eq!(server.get_active_connections_count(), 0);
    }

    #[test]
    fn test_no_tokio_dependency() {
        // 这个测试验证协议层没有Tokio依赖
        // 所有操作都是同步的，可以在标准线程中运行
        
        let storage_dispatcher = SingleThreadCommandDispatcher::new();
        let mut handler = DecoupledConnectionHandler::new(
            1,
            Arc::new(Mutex::new(storage_dispatcher)),
        );

        // 在标准线程中处理（不需要Tokio运行时）
        let data = b"*2\r\n$3\r\nGET\r\n$3\r\nkey\r\n".to_vec();
        assert!(handler.handle_data(data).is_ok());

        // 同步等待和检查结果
        std::thread::sleep(Duration::from_millis(20));
        
        match handler.get_response_timeout(Duration::from_millis(100)) {
            Some(ConnectionResponse::SendData(_)) => {
                println!("✅ 协议层成功处理命令（无Tokio依赖）");
            }
            _ => {
                println!("⚠️ 未收到预期响应");
            }
        }

        assert!(handler.stop().is_ok());
    }
}